package hz

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}
import org.elasticsearch.spark._

import scala.util.control._
import scala.collection.mutable.ArrayBuffer
/**
  * Created by hz on 2017/1/6.
  */
object Kafkas {

  def main(args:Array[String]){
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Setting.readProperties("default1.properties")
    val sparkConf = new SparkConf().set("spark.cores.max","1")//.setAppName("test")
    //val sparkConf = new SparkConf().setMaster("local[2]").setAppName("test")
    sparkConf.set("es.nodes", Setting.getProperty("esnode"))//spark://hz:7077  ElasticSearch
    sparkConf.set("es.port", Setting.getProperty("esport"))
    sparkConf.set("es.index.auto.create", "true")
    val scc = new StreamingContext(sparkConf, Duration(5000))
    val topics = Set(Setting.getProperty("topic"))
    val kafkaParam = Map(
      "metadata.broker.list" -> Setting.getProperty("brokerlist") // kafka的broker list地址
    )
    val b=new SimpleDateFormat("yyyy-MM-dd")
    val stream: InputDStream[(String, String)] = createStream(scc, kafkaParam, topics)
    val lei = stream.map(_._2)
    lei.persist()

    lei.print()
    val rule=DBUtil.getRule(args(0))
    val min1:Int=min(rule);
    val h = lei.window(Seconds(min1), Seconds(5))

    if (rule.size() == 0) {
      lei.print()
    } else {
      val loop=new Breaks;
      lei.print()
      var rs=h.foreachRDD(rdd => {
        loop.breakable{
          for (i <- 0 until rule.size()) {
            println("普通规则")
            val rs=zhixing(rdd, rule.get(i))
            if (rs.count()>0){
              rs.saveToEs("baojing1"+b.format(new Date())+"/guizefenlei")
            }
          }
          if (rule.size()==1){
            println("场景1")
            var rs=zhixing(rdd,rule.get(0),args(0))
            if (rs.count()>0){
              rs.saveToEs("baojingchangjing"+b.format(new Date())+"/guizefenlei")
            }
          }
          if (rule.size()>1){
            println("场景2")
            var i=0;
            var rs=zhixing(rdd,rule.get(0),args(0))
            while(rs.count()>0 && i<rule.size()){
              rs=zhixing1(rs,rule.get(i),args(0))
              i+=1
            }
            if (rs.count()>0 && i==rule.size()-1){
              rs.saveToEs("baojingchangjing"+b.format(new Date())+"/guizefenlei")
            }
          }
        }
      })


    }


    scc.start() // 真正启动程序
    scc.awaitTermination() //阻塞等待
  }


  def min(rules:util.List[RuleBean]):Int={
    var min=rules.get(0).getTimerange;
    for (i <-0 until rules.size()){
      if (min>rules.get(i).getTimerange) min=rules.get(i).getTimerange
    }
    min
  }



  def zhixing(lei:RDD[String],rule:RuleBean,cj:String=""):RDD[Map[String,String]]={
    val leixing:String=DBUtil.getLeixing
    var rizhi1:RDD[Map[String,String]]=null

    println("这是第"+rule.getId+"")
    val rizhi:RDD[Map[String,String]]=lei.map(x => denghao(x)).map(line => jiexi(line, rule.getId,cj)).filter(_.contains(leixing))

    rizhi1=tiaojian(rizhi,rule,cj)

    val rs=paixu1(rizhi1,rule)

    rs
  }


  //场景
  def zhixing1(lei:RDD[Map[String,String]],rule:RuleBean,cj:String=""):RDD[Map[String,String]]={

    var rizhi1:RDD[Map[String,String]]=null
    rizhi1=tiaojian(lei,rule,cj)

    val rs=paixu1(rizhi1,rule)

    rs
  }

  /**
    * 进行条件匹配
    *
    *
    * @param rule1
    * @return
    */
  def tiaojian(rizhi:RDD[Map[String,String]],rule1:RuleBean,cj:String):RDD[Map[String,String]]={
    var rizhi1=rizhi

    val size=rule1.getTiaojian().size()
    val tiaojian= rule1.getTiaojian();
    if (tiaojian.size()!=0){
      for (i <- 0 until size) {
        val ziduan=tiaojian.get(i).getZiduan
        val zhi=tiaojian.get(i).getZhi
        tiaojian.get(i).getCaozuofu match {
          case "=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 == zhi).map(x => x._1)
          case "<" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 < zhi).map(x => x._1)
          case ">" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 > zhi).map(x => x._1)
          case "<=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 <= zhi).map(x => x._1)
          case ">=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 >= zhi).map(x => x._1)
          case "!=" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2 != zhi).map(x => x._1)
          case "Contains" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.contains(zhi)).map(x => x._1)
          case "StartWith" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.startsWith(zhi)).map(x => x._1)
          case "EndWith" =>rizhi1 = rizhi1.map(x => (x, x(ziduan))).filter(_._2.endsWith(zhi)).map(x => x._1)
        }
      }
    }
    rizhi1
  }


  def denghao(list:String):Array[String]={
    var tmp=""
    var ab=ArrayBuffer[String]()
    val length=list.length
    val exp="""([\w]+)=([\w]+[-(\+]?[\S]+)""".r
    val re=exp.findAllIn(list)
    for (i <- re){
      ab+=i
    }

    val exp1="""([\w]+)=\\\"([^\"]*)\\\"""".r
    val re1=exp1.findAllIn(list)
    for (j <- re1){
      print("woshijinlaile")
      ab+=j
    }
    val rab=ab.toArray
    for (i <- ab){
      print(i+" ")
    }
    return rab
  }


  /**
    * 进行排序
    *
    * @param list
    * @param rule
    */
  def paixu1(list:RDD[Map[String,String]],rule:RuleBean):RDD[Map[String,String]]={
    println("==进入排序==")
    var hh=list
    hh.cache()

    /**
      * 相同聚合时
      *
      * @param hello
      * @return
      */
    def zuhea(hello:Map[String,String]):String={
      val list=rule.getSamepol.split(",")
      var rs:String=""
      for (i <- list){
        rs+=hello(i)
      }
      rs
    }

    /**
      * 不同时聚合
      *
      * @param hello
      * @return
      */
    def zuheb(hello:Map[String,String]):String={
      val list=rule.getDiffpol.split(",")
      var rs:String=""
      for (i <- list){
        rs+=hello(i)
      }
      rs
    }




    var gv1:RDD[Iterable[Map[String,String]]]=null

    gv1=hh.map(x=>(zuhea(x),x)).groupByKey().map(_._2)
    val size=rule.getNumber

    val daxiao:RDD[Map[String,String]]= gv1.map(x=>(x,x.map(y=>zuheb(y)).toArray.distinct.size)).filter(_._2>=size).flatMap(x=>x._1.map(y=>y))  //不同时聚合




    /* if (daxiao.count() > 0)  // RDD has data
     {
       val b=new SimpleDateFormat("yyyy-MM-dd")
       println("存进去了")*/
    //daxiao.saveToEs("baojing"+b.format(new Date())+"/guizefenlei")
    //SpringUtil.getInstance().setbaojing()
    /* jingtai.baojing()
     val ip:String=daxiao.take(1).apply(0)("dst")
     val id=SpringUtil.getInstance().getasset(ip)

     val warn=new WarningBean();
     warn.setWarnname(rule.getRulename)
     warn.setAssetid(id)
     warn.setStatus("未清除")
     warn.setType("事件告警")
     warn.setRuntime((new java.util.Date().getTime-rule.getMktime.getTime).toInt)
     warn.setRuleid(rule.getId)
     //SpringUtil.getInstance().savebaojing(warn);
     warn11.save(warn)
     daxiao.count()*/

    //}
    daxiao
  }





  /**
    * 创建一个从kafka获取数据的流.
    *
    * @param scc           spark streaming上下文
    * @param kafkaParam    kafka相关配置
    * @param topics        需要消费的topic集合
    * @return
    */
  def createStream(scc: StreamingContext, kafkaParam: Map[String, String], topics: Set[String]) = {
    KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](scc, kafkaParam, topics)
  }

  /**
    * 将字符串解析为Map
    *
    * @param map
    * @param guize1
    * @return
    */
  def jiexi(map:Array[String],guize1:Int,cj:String):Map[String,String]={
    var map1:Map[String,String]=Map();
    var guize:Int=0;
    for (i <- map.indices){
      var cache1,cache2=""
      cache1=map(i).split("=")(0)
      cache2=map(i).split("=")(1)
      map1+=(cache1 -> cache2.replaceAll("""\\""","").replaceAll("\"",""))
    }
    val a=Array("dst","src")
    for(i <- 0 until a.size if !map1.contains(a(i))) map1+=(a(i) -> " ")
    map1+=("guize"->guize1.toString)
    if (cj!=""){ map1+=("cj"->cj)}

    map1
  }

}
